package com.zhang.spark_2.com.zhang.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @title: 页面单跳转换率统计(统计指定列表页面)
 * @author: zhang
 * @date: 2022/2/16 17:16 
 */
object Spark03_req_1 {

  def main(args: Array[String]): Unit = {
    //TODO 获取环境
    val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(conf);
    val userVisit: RDD[String] = sc.textFile("data/user_visit_action.txt")

    val mapObjs: RDD[UserVisitAction] = userVisit.map(
      line => {
        val data: Array[String] = line.split("_")
        UserVisitAction(
          data(0),
          data(1).toLong,
          data(2),
          data(3).toLong,
          data(4),
          data(5),
          data(6).toLong,
          data(7).toLong,
          data(8),
          data(9),
          data(10),
          data(11),
          data(12).toLong
        )
      }
    )

    mapObjs.cache()
    val list1 = List(1, 2, 3, 4, 5, 6,7)
    val okPageIds: List[(Int, Int)] = list1.zip(list1.tail)

    //todo 计算分母
    val pageIds: Map[Long, Int] = mapObjs
      .filter(
        obj => {
          list1.contains(obj.page_id)
        }
      )
      .map(
        obj => (obj.page_id, 1)
      )
      .reduceByKey(_ + _)
      .collect()
      .toMap



    //todo 计算分子
    val pageFlow: RDD[((Long, Long), Int)] = mapObjs
      .groupBy(_.session_id)
      .mapValues(
        iter => {
          val actions: List[UserVisitAction] = iter.toList.sortBy(_.action_time)
          val pageIds: List[Long] = actions.map(_.page_id)
          val tuples: List[(Long, Long)] = pageIds.zip(pageIds.tail)
          tuples
            .filter(
              t=>okPageIds.contains(t)
            )
            .map(
              t => {
                ((t._1, t._2), 1)
              }
            )
        }
      )
      .map(_._2)
      .flatMap(list => list)
      .reduceByKey(_ + _)


    //todo 单跳转换率
    pageFlow.foreach {
      case ((p1, p2), cnt) => {
        val count: Int = pageIds.getOrElse(p1, 0)
        println(s"页面$p1 ---> $p2:转化率为：" + (cnt.toDouble  / count))
      }
    }


    sc.stop()
  }


  //用户访问动作表
  case class UserVisitAction(
                              date: String, //用户点击行为的日期
                              user_id: Long, //用户的ID
                              session_id: String, //Session的ID
                              page_id: Long, //某个页面的ID
                              action_time: String, //动作的时间点
                              search_keyword: String, //用户搜索的关键词
                              click_category_id: Long, //某一个商品品类的ID
                              click_product_id: Long, //某一个商品的ID
                              order_category_ids: String, //一次订单中所有品类的ID集合
                              order_product_ids: String, //一次订单中所有商品的ID集合
                              pay_category_ids: String, //一次支付中所有品类的ID集合
                              pay_product_ids: String, //一次支付中所有商品的ID集合
                              city_id: Long //城市 id
                            )

}
